package storm.bolt;

import org.apache.storm.trident.TridentState;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.operation.builtin.Count;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.trident.testing.MemoryMapState;
import org.apache.storm.trident.testing.Split;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

/**
 * Created by luoqifei on 2016/10/12.
 */
//storm Trident 为这些小块提供了一个完全成熟的批处理 API。
// 这个 API 和你见到过的 Pig 或者 Cascading 这样的 Hadoop 的高级抽象语言很相似：
// 你可以处理分组（group by）、联结（join）、聚合（aggregation）、函数（function）、过滤器（filter）等各种操作

public class StormFuntionDemo extends BaseFunction {
    public void execute(TridentTuple tuple, TridentCollector collector) {
        for(int i=0; i < tuple.getInteger(0); i++) {
            //元组中大于第一个数的，都可以进行发射回到流中，其他的就不处理，相当于被过滤掉，元组回不到流了。
            collector.emit(new Values(i));
        }
    }
    public static void main(String args[]){
        //数据源
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 3,
                new Values("the cow jumped over the moon"),
                new Values("the man went to the store and bought some candy"),
                new Values("four score and seven years ago"),
                new Values("how many apples can you eat"));
        spout.setCycle(true);
        //可顶一个一个整型数组数据源 intSpout
        //统计单词
        TridentTopology topology = new TridentTopology();
        TridentState wordCounts =
                topology.newStream("spout1", spout)
                        .each(new Fields("sentence"), new Split(), new Fields("word"))
                        // .each(new Fields("number"), new StormFuntionDemo(), new Fields("zz"))
                        .groupBy(new Fields("word"))
                        .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"))
                        .parallelismHint(6);

    }
}
